package cn.doitedu.etl;

import cn.doitedu.utils.GeohashUDF;
import cn.doitedu.utils.Map2JsonUDF;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/1/14
 * @Desc: 学大数据，上多易教育
 *   app埋点日志处理： 从ods加工到dwd
 *   处理需求：过滤、清洗、字段统
 *           公共维度退维！
 **/
public class Job01_AppEvents_ODS_DWD {

    public static void main(String[] args) {
        Configuration conf = new Configuration();
        //conf.setInteger("rest.port",8089);
        conf.setString(RestOptions.BIND_PORT, "8081-8089");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);


        // 编程环境
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射 kafka 的ods层的行日志topic: ods-events
        tenv.executeSql(
                " create table ods_events_kafka(                               "+
                        "   release_channel string,                            "+
                        " 	device_type string,                                "+
                        " 	session_id string,                                 "+
                        " 	lat double,                                        "+
                        " 	lng double,                                        "+
                        " 	user_name string,                                  "+
                        " 	event_id string,                                   "+
                        " 	action_time bigint,                                "+
                        " 	properties map<string,string>,                     "+
                        " 	pt as proctime()                                   "+
                        " ) WITH (                                             "+
                        "   'connector' = 'kafka',                             "+
                        "   'topic' = 'ods-events',                            "+
                        "   'properties.bootstrap.servers' = 'doitedu:9092',   "+
                        "   'properties.group.id' = 'doit44_g1',               "+
                        "   'scan.startup.mode' = 'latest-offset',             "+
                        "   'value.format' = 'json',                           "+
                        "   'value.fields-include' = 'EXCEPT_KEY'              "+
                        " )                                                    "
        );



        // 建表，映射 hbase中 用户注册信息维表:  dim_user_info
        tenv.executeSql(
                " create table dim_user_info_hbase(                  "+
                        "      user_name string                         "+
                        " 	   ,f  ROW<                             " +
                        "            id bigint                       "+
                        "            ,member_level_id bigint           "+
                        "            ,password string                  "+
                        "            ,nickname string                  "+
                        "            ,phone string                     "+
                        "            ,status int                       "+
                        "            ,create_time timestamp(3)         "+
                        "            ,icon string                      "+
                        "            ,gender int                       "+
                        "            ,birthday date                    "+
                        "            ,city string                      "+
                        "            ,job string                       "+
                        "            ,personalized_signature string    "+
                        "            ,source_type int                  "+
                        "            ,integration int                  "+
                        "            ,growth int                       "+
                        "            ,luckey_count int                 "+
                        "            ,history_integration int          "+
                        "            ,modify_time timestamp(3)>        "+
                        "    ,primary key(user_name) not enforced      "+
                        " ) WITH (                                     "+
                        "  'connector' = 'hbase-2.2',                  "+
                        "  'table-name' = 'dim_user_info',             "+
                        "  'zookeeper.quorum' = 'doitedu:2181'         "+
                        " )                                            "
        );





        // 建表，映射 hbase中 地理位置维表:  dim_geo_area
        tenv.executeSql(
                " create table dim_geo_area_hbase(                     "+
                        "    geohash  string,                                  "+
                        "    f row<province string,city string,region string>, "+
                        "    primary key(geohash) not enforced                 "+
                        " ) WITH (                                             "+
                        "  'connector' = 'hbase-2.2',                          "+
                        "  'table-name' = 'dim_geo_area',                      "+
                        "  'zookeeper.quorum' = 'doitedu:2181',                "+
                        "  'lookup.cache' = 'PARTIAL',                         "+
                        "  'lookup.partial-cache.max-rows' = '100000',         "+
                        "  'lookup.partial-cache.expire-after-access' = '10hour' "+
                        " )                                                    "
        );




        // 建表，映射 hbase中 页面信息维表： dim_page_info
        tenv.executeSql(
                " create table dim_page_info_hbase(                               "+
                        "    url_prefix  string,                                  "+
                        "    f row<page_type string,page_service string>,         "+
                        "    primary key(url_prefix) not enforced                 "+
                        " ) WITH (                                                "+
                        "  'connector' = 'hbase-2.2',                             "+
                        "  'table-name' = 'dim_page_info',                        "+
                        "  'zookeeper.quorum' = 'doitedu:2181',                   "+
                        "  'lookup.cache' = 'PARTIAL',                            "+
                        "  'lookup.partial-cache.max-rows' = '100000',            "+
                        "  'lookup.partial-cache.expire-after-access' = '10hour', "+
                        "  'lookup.async' = 'true'                                "+
                        " )                                                    "
        );



        // 建表，映射 kafka 的dwd层的topic: dwd-events
        tenv.executeSql("create table dwd_events_kafka(\n" +
                "     release_channel    string\n" +
                "    ,device_type        string\n" +
                "    ,session_id         string\n" +
                "    ,lat                double\n" +
                "    ,lng                double\n" +
                "    ,user_name          string\n" +
                "    ,event_id           string\n" +
                "    ,action_time        bigint\n" +
                "    ,properties         map<string,string>\n" +
                "    ,user_id bigint\n" +
                "    ,member_level_id bigint\n" +
                "    ,nickname string\n" +
                "    ,phone string\n" +
                "    ,status int\n" +
                "    ,create_time timestamp(3)\n" +
                "    ,icon string\n" +
                "    ,gender int\n" +
                "    ,birthday date\n" +
                "    ,register_city string\n" +
                "    ,job string\n" +
                "    ,personalized_signature string\n" +
                "    ,source_type int\n" +
                "    ,integration int\n" +
                "    ,growth int\n" +
                "    ,luckey_count int\n" +
                "    ,history_integration int\n" +
                "    ,modify_time timestamp(3)\n" +
                "    ,province   string\n" +
                "    ,city       string\n" +
                "    ,region     string\n" +
                "    ,page_type       string\n" +
                "    ,page_service    string\n" +
                "\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'dwd-events',\n" +
                "    'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "    'properties.group.id' = 'doit44_g1',\n" +
                "    'scan.startup.mode' = 'latest-offset',\n" +
                "    'value.format' = 'json',\n" +
                "    'value.fields-include' = 'EXCEPT_KEY'\n" +
                ")                                                    ");

        // 创建doris结果表的连接器映射表
        tenv.executeSql(
                " create table user_events_detail_doris(                            "+
                        "     event_id                   varchar(100)               "+
                        "     ,province                  varchar(100)               "+
                        "     ,city                      varchar(100)               "+
                        "     ,region                    varchar(100)	            "+
                        "     ,dt                        date                       "+
                        "     ,release_channel           varchar(100)               "+
                        "     ,device_type               varchar(100)               "+
                        "     ,session_id                varchar(100)               "+
                        "     ,lat                       double                     "+
                        "     ,lng                       double                     "+
                        "     ,user_name                 varchar(100)               "+
                        "     ,action_time               bigint                     "+
                        "     ,properties_json           varchar(100)               "+
                        "     ,user_id                   bigint                     "+
                        "     ,member_level_id           int                        "+
                        "     ,nickname                  varchar(100)               "+
                        "     ,phone                     varchar(100)               "+
                        "     ,status                    int                        "+
                        "     ,create_time               timestamp(3)               "+
                        "     ,icon                      varchar(100)               "+
                        "     ,gender                    int                        "+
                        "     ,birthday                  date                       "+
                        "     ,register_city             varchar(100)               "+
                        "     ,job                       varchar(100)               "+
                        "     ,personalized_signature    varchar(100)               "+
                        "     ,source_type               int                        "+
                        "     ,integration               int                        "+
                        "     ,growth                    int                        "+
                        "     ,luckey_count              int                        "+
                        "     ,history_integration       int                        "+
                        "     ,modify_time               timestamp(3)               "+
                        "     ,page_type                 varchar(100)               "+
                        "     ,page_service              varchar(100)               "+
                        " ) WITH (                                                  "+
                        "    'connector' = 'doris',                                 "+
                        "    'fenodes' = 'doitedu:8030',                            "+
                        "    'table.identifier' = 'dwd.user_events_detail',         "+
                        "    'username' = 'root',                                   "+
                        "    'password' = 'root',                                   "+
                        "    'sink.label-prefix' = 'doris_label-007'                "+
                        " )                                                         "

        );




        // lookup join
        tenv.createTemporaryFunction("geo", GeohashUDF.class);
        //tenv.executeSql("select regexp_extract('/ab/xxxxx/cd/a.html?9sdg','(/.*/).*?')").print();
        tenv.executeSql(
                "create temporary view joined_events AS  \n" +
                "select /*+ LOOKUP('table'='dim_user_info_hbase', 'retry-predicate'='lookup_miss', 'retry-strategy'='fixed_delay', 'fixed-delay'='100ms','max-attempts'='3') */  \n" +
                "a.release_channel\n" +
                ",a.device_type\n" +
                ",a.session_id\n" +
                ",a.lat\n" +
                ",a.lng\n" +
                ",a.user_name\n" +
                ",a.event_id\n" +
                ",a.action_time\n" +
                ",a.properties\n" +
                ",u.f.id as user_id \n" +
                ",u.f.member_level_id\n" +
                ",u.f.nickname\n" +
                ",u.f.phone\n" +
                ",u.f.status\n" +
                ",u.f.create_time\n" +
                ",u.f.icon\n" +
                ",u.f.gender\n" +
                ",u.f.birthday\n" +
                ",u.f.city as register_city\n" +
                ",u.f.job\n" +
                ",u.f.personalized_signature\n" +
                ",u.f.source_type\n" +
                ",u.f.integration\n" +
                ",u.f.growth\n" +
                ",u.f.luckey_count\n" +
                ",u.f.history_integration\n" +
                ",u.f.modify_time\n" +
                ",g.f.province\n" +
                ",g.f.city\n" +
                ",g.f.region\n" +
                ",p.f.page_type\n" +
                ",p.f.page_service\n" +
                "from ods_events_kafka a\n" +
                "LEFT JOIN dim_user_info_hbase for system_time as of a.pt as u ON a.user_name=u.user_name\n" +
                "LEFT JOIN dim_geo_area_hbase for system_time as of  a.pt as g ON geo(a.lat,a.lng) = g.geohash\n" +
                "LEFT JOIN dim_page_info_hbase for system_time as of a.pt as p ON REGEXP_EXTRACT(a.properties['url'],'(/.*/).*?' )= p.url_prefix\n");


        // 将join后的结果写入kafka表
        tenv.executeSql("insert into dwd_events_kafka select * from joined_events");

        // 将join后的结果写入doris表
        tenv.createTemporaryFunction("map_json", Map2JsonUDF.class);
        tenv.executeSql(
                " INSERT INTO user_events_detail_doris                                                      "+
                        " SELECT                                                                            "+
                        "     event_id                                                                      "+
                        "     ,province                                                                     "+
                        "     ,city                                                                         "+
                        "     ,region                                                                       "+
                        "     ,to_date(date_format(to_timestamp_ltz(action_time,3),'yyyy-MM-dd')) as  dt    "+
                        "     ,release_channel                                                              "+
                        "     ,device_type                                                                  "+
                        "     ,session_id                                                                   "+
                        "     ,lat                                                                          "+
                        "     ,lng                                                                          "+
                        "     ,user_name                                                                    "+
                        "     ,action_time                                                                  "+
                        "     ,map_json(properties) as properties_json                                      "+
                        "     ,user_id                                                                      "+
                        "     ,cast(member_level_id as int) as  member_level_id                             "+
                        "     ,nickname                                                                     "+
                        "     ,phone                                                                        "+
                        "     ,status                                                                       "+
                        "     ,create_time                                                                  "+
                        "     ,icon                                                                         "+
                        "     ,gender                                                                       "+
                        "     ,birthday                                                                     "+
                        "     ,register_city                                                                "+
                        "     ,job                                                                          "+
                        "     ,personalized_signature                                                       "+
                        "     ,source_type                                                                  "+
                        "     ,integration                                                                  "+
                        "     ,growth                                                                       "+
                        "     ,luckey_count                                                                 "+
                        "     ,history_integration                                                          "+
                        "     ,modify_time                                                                  "+
                        "     ,page_type                                                                    "+
                        "     ,page_service                                                                 "+
                        " FROM joined_events                                                                "
        );
    }

}
